package com.fudan.run.trans;

import java.util.HashMap;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;

import org.apache.spark.api.java.JavaRDD;

import com.fudan.cfg.trans.FormatFieldTransDef;
import com.fudan.run.JobRunner;
import com.fudan.run.VariableRddMap;
import com.fudan.run.ctx.annotation.TransAdapter;

@TransAdapter("format_field")
public class FormatFieldTrans extends TransHandler<FormatFieldTransDef> {

	@Override
	public void trans(JobRunner jobRunner, FormatFieldTransDef transNode, VariableRddMap variableMap) {
		JavaRDD<Map<String, Object>> targetRdd = variableMap.getRdd(transNode.getTarget());
		JavaRDD<Map<String, Object>> resultRdd = targetRdd.map(record->{
			List<Map<String,Object>> mappingList = transNode.getMapping();
			for (Map<String, Object> mapping : mappingList) {
				String fieldName = (String)mapping.get("field");
				Object value = record.get(fieldName);
				String type = (String)mapping.get("type");
				switch (type) {
				case "map": {
					if(value != null && value instanceof Map) {
						List<Map<String,Object>> list = new LinkedList<>();
						Set<Entry<String, Object>> valueEntry = ((Map<String,Object>)value).entrySet();
						Map<String,Object> formatDef = (Map<String, Object>) mapping.get("format");
						
						for(Entry<String,Object> ve :valueEntry) {
							String kp = (String) formatDef.get("key");
							String vp = (String) formatDef.get("value");
							Map<String,Object> v = new HashMap<>();
							v.put(kp, ve.getKey());
							v.put(vp, ve.getValue());
//							Set<String> newKeySet = formatDef.keySet();
//							for (String newKey : newKeySet) {
//								Object nv = formatDef.get(newKey);
//								if("%key".equals(nv)) {
//									nv = ve.getKey();
//								}else if("%value".equals(nv)) {
//									nv = ve.getValue();
//								}
//								v.put(newKey, nv);
//							}
							list.add(v);
						}
						record.put(fieldName, list);
					}
					break;
				}
					

				default:
					break;
				}
			}
			
			
			return record;
		});
		variableMap.addRdd(transNode.getName(), resultRdd);
	}

}
